package com.king.first.app.ad;

import com.king.first.bean.MarketingUserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;

import java.util.*;

public class Project_AppAnalysis_By_Chanel {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<MarketingUserBehavior> streamSource = env.addSource(new AppMarketingDataSource());

        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = streamSource.map(new MapFunction<MarketingUserBehavior, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(MarketingUserBehavior marketingUserBehavior) throws Exception {
                return Tuple2.of(marketingUserBehavior.getChannel() + "_" + marketingUserBehavior.getBehavior(), 1);
            }
        });

        SingleOutputStreamOperator<Map<String, Integer>> process = streamOperator.keyBy(k -> k.f0).process(new MyKeyedProcessFunction());

        process.print();
//        streamOperator.keyBy(k -> k.f0).sum(1).print();

        env.execute();
    }

    public static class MyKeyedProcessFunction extends KeyedProcessFunction<String, Tuple2<String, Integer>, Map<String, Integer>>{

        Map<String, Integer> channelCount = new HashMap<>();

        @Override
        public void processElement(Tuple2<String, Integer> value,
                                   KeyedProcessFunction<String, Tuple2<String, Integer>,
                                           Map<String, Integer>>.Context context,
                                   Collector<Map<String, Integer>> collector) throws Exception {
            Integer count = channelCount.getOrDefault(value.f0, 0);
            count ++;
            channelCount.put(value.f0, count);
            collector.collect(channelCount);
        }
    }

    public static class AppMarketingDataSource extends RichSourceFunction<MarketingUserBehavior> {
        boolean canRun = true;
        Random random = new Random();
        List<String> channels = Arrays.asList("huawwei", "xiaomi", "apple", "baidu", "qq", "oppo", "vivo");
        List<String> behaviors = Arrays.asList("download", "install", "update", "uninstall");

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (canRun) {
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        (long) random.nextInt(1000000),
                        behaviors.get(random.nextInt(behaviors.size())),
                        channels.get(random.nextInt(channels.size())),
                        System.currentTimeMillis());
                ctx.collect(marketingUserBehavior);
                Thread.sleep(500);
            }
        }

        @Override
        public void cancel() {
            canRun = false;
        }
    }
}

